#include <syncupconfig.h>
#include <async_queue.h>
#include <errno.h>
#include <time.h>
#ifdef HAVE_SYS_TIME_H
#include <sys/time.h>
#endif
#include <glib.h>


struct async_queue_entry {
    void *data;
    struct list_head queue;
};

void
async_queue_init(struct async_queue *queue)
{
    INIT_LIST_HEAD(&queue->head);
    pthread_mutex_init(&queue->mutex, NULL);
    pthread_cond_init(&queue->cond, NULL);
    queue->waiter_n = 0;
    queue->interrupted = FALSE;
}

void
async_queue_free(struct async_queue *queue)
{
    struct list_head *pos, *t;
    struct list_head *head = &queue->head;

    list_for_each_safe(pos, t, head) {
        list_del(pos);
        g_slice_free(struct async_queue_entry,
                list_entry(pos, struct async_queue_entry, queue));
    }
    pthread_mutex_destroy(&queue->mutex);
    pthread_cond_destroy(&queue->cond);
    queue->waiter_n = 0;
    queue->interrupted = FALSE;
}

void
async_queue_push(struct async_queue *queue, void *data)
{
    struct async_queue_entry *ent = g_slice_new(struct async_queue_entry);
    ent->data = data;
    pthread_mutex_lock(&queue->mutex);
    list_add_tail(&ent->queue, &queue->head);

    /* Signal condition inside critical region to ensure that
     * async_queue_timeout_pop() won't get timeout when there are data pushed
     * into queue.
     */
    pthread_cond_signal(&queue->cond);

    pthread_mutex_unlock(&queue->mutex);
}

void *
async_queue_pop(struct async_queue *queue)
{
    struct list_head *first;
    struct async_queue_entry *ent;
    void *data;

    pthread_mutex_lock(&queue->mutex);
    ++queue->waiter_n;
    while (!queue->interrupted && list_empty(&queue->head)) {
        pthread_cond_wait(&queue->cond, &queue->mutex);
    }
    --queue->waiter_n;
    if (queue->interrupted) {
        if (queue->waiter_n == 0) {
            queue->interrupted = FALSE;
        }
        pthread_mutex_unlock(&queue->mutex);
        return NULL;
    }
    first = queue->head.next;
    list_del(first);
    pthread_mutex_unlock(&queue->mutex);

    ent = list_entry(first, struct async_queue_entry, queue);
    data = ent->data;
    g_slice_free(struct async_queue_entry, ent);
    return data;
}


static inline void
get_abs_time(struct timespec *ts)
{
#ifdef HAVE_GETTIMEOFDAY
	struct timeval tv;
	gettimeofday(&tv, NULL);
	ts->tv_sec = tv.tv_sec;
	ts->tv_nsec = tv.tv_usec * 1000;
#else
	ts->tv_sec = time(NULL);
	ts->tv_nsec = 0;
#endif
}

void *
async_queue_timeout_pop(struct async_queue *queue,
        const struct timespec *timeout)
{
    struct list_head *first;
    struct async_queue_entry *ent;
    void *data;
    int err;
    struct timespec ts;

#ifdef HAVE_CLOCK_GETTIME
    clock_gettime(CLOCK_REALTIME, &ts);
#else
    get_abs_time(&ts);
#endif
    ts.tv_nsec += timeout->tv_nsec;
    ts.tv_sec += timeout->tv_sec;
    if (ts.tv_nsec > 1000000000) {
        ts.tv_sec += ts.tv_nsec / 1000000000;
        ts.tv_nsec = ts.tv_nsec % 1000000000;
    }


    err = 0;
    pthread_mutex_lock(&queue->mutex);
    ++queue->waiter_n;
    while (!queue->interrupted && list_empty(&queue->head)) {
        err = pthread_cond_timedwait(&queue->cond, &queue->mutex, &ts);
        if (err) {
            break;
        }
    }
    --queue->waiter_n;
    if (err) {
        goto err_ret;
    }
    if (queue->interrupted) {
        if (queue->waiter_n == 0) {
            queue->interrupted = FALSE;
        }
        goto err_ret;
    }
    first = queue->head.next;
    list_del(first);
    pthread_mutex_unlock(&queue->mutex);

    ent = list_entry(first, struct async_queue_entry, queue);
    data = ent->data;
    g_slice_free(struct async_queue_entry, ent);
    return data;

err_ret:
    pthread_mutex_unlock(&queue->mutex);
    return NULL;
}

void
async_queue_interrupt(struct async_queue *queue)
{
    pthread_mutex_lock(&queue->mutex);
    queue->interrupted = TRUE;
    pthread_cond_broadcast(&queue->cond);
    pthread_mutex_unlock(&queue->mutex);
}

void
async_queue_lock(struct async_queue *queue)
{
    pthread_mutex_lock(&queue->mutex);
}

void
async_queue_unlock(struct async_queue *queue)
{
    pthread_mutex_unlock(&queue->mutex);
}

void
async_queue_push_unlocked(struct async_queue *queue, void *data)
{
    struct async_queue_entry *ent = g_slice_new(struct async_queue_entry);
    ent->data = data;
    list_add_tail(&ent->queue, &queue->head);
    pthread_cond_signal(&queue->cond);
}

void *
async_queue_pop_unlocked(struct async_queue *queue)
{
    struct list_head *first;
    struct async_queue_entry *ent;
    void *data;

    ++queue->waiter_n;
    while (!queue->interrupted && list_empty(&queue->head)) {
        pthread_cond_wait(&queue->cond, &queue->mutex);
    }
    --queue->waiter_n;
    if (queue->interrupted) {
        if (queue->waiter_n == 0) {
            queue->interrupted = FALSE;
        }
        return NULL;
    }
    first = queue->head.next;
    list_del(first);

    ent = list_entry(first, struct async_queue_entry, queue);
    data = ent->data;
    g_slice_free(struct async_queue_entry, ent);
    return data;
}

void *
async_queue_timeout_pop_unlocked(struct async_queue *queue,
        const struct timespec *timeout)
{
    struct list_head *first;
    struct async_queue_entry *ent;
    void *data;
    int err;
    struct timespec ts;

#ifdef HAVE_CLOCK_GETTIME
    clock_gettime(CLOCK_REALTIME, &ts);
#else
    get_abs_time(&ts);
#endif
    ts.tv_nsec += timeout->tv_nsec;
    ts.tv_sec += timeout->tv_sec;
    if (ts.tv_nsec > 1000000000) {
        ts.tv_sec += ts.tv_nsec / 1000000000;
        ts.tv_nsec = ts.tv_nsec % 1000000000;
    }

    err = 0;
    ++queue->waiter_n;
    while (!queue->interrupted && list_empty(&queue->head)) {
        err = pthread_cond_timedwait(&queue->cond, &queue->mutex, &ts);
        if (err) {
            break;
        }
    }
    --queue->waiter_n;
    if (err) {
        return NULL;
    }
    if (queue->interrupted) {
        if (queue->waiter_n == 0) {
            queue->interrupted = FALSE;
        }
        return NULL;
    }
    first = queue->head.next;
    list_del(first);

    ent = list_entry(first, struct async_queue_entry, queue);
    data = ent->data;
    g_slice_free(struct async_queue_entry, ent);
    return data;
}

void
async_queue_interrupt_unlocked(struct async_queue *queue)
{
    queue->interrupted = TRUE;
    pthread_cond_broadcast(&queue->cond);
}
